package com.bd08.flink.demo

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time


object WindowSWordCount  extends App {

  case class WordCount(word:String,count:Int)


  val env = StreamExecutionEnvironment.getExecutionEnvironment
  //val ds = env.addSource(new SocketTextStreamFunction("172.18.172.150",9999,"\n",3))
  val ds = env.socketTextStream("172.18.169.160",9999,'\n')
    //.assignAscendingTimestamps(line => line.hashCode)

    .flatMap(line => line.split("\\s")).map(WordCount(_,1)).keyBy(wc => wc.word)

    //.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(2))).sum("count")
    //.countWindow(5).sum("count")
    .countWindow(5,2).sum("count")

  ds.print()
  env.execute("window-s-word-count")


}
